package cn.learn.server;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;

public class NettyServerCnxn {
    private final Channel channel;
    final ZooKeeperServer zkServer;
    private long sessionId;
    private final NettyServerCnxnFactory factory;
    private ByteBuffer bb;
    private final ByteBuffer bbLen = ByteBuffer.allocate(4);

    public NettyServerCnxn(Channel channel, ZooKeeperServer zks, NettyServerCnxnFactory factory) {
        this.channel = channel;
        this.zkServer = zks;
        this.factory = factory;
    }

    public void processMessage(ByteBuf buf) throws IOException {
        System.out.println("通过netty获得的数据=》" + buf.toString());
        receiveMessage(buf);
    }

    private void receiveMessage(ByteBuf message) throws IOException {
        while (message.isReadable()) {
            if (bb != null) {
                if (bb.remaining() > message.readableBytes()) {
                    int newLimit = bb.position() + message.readableBytes();
                    bb.limit(newLimit);
                }
                message.readBytes(bb);
                bb.limit(bb.capacity());

                if (bb.remaining() == 0) {
                    bb.flip();
                    System.out.println(bb.toString());
                    ZooKeeperServer zks = this.zkServer;
                    zks.processPacket(this, bb);
                }
                bb = null;
            } else {
                if (message.readableBytes() < bbLen.remaining()) {
                    bbLen.limit(bbLen.position() + message.readableBytes());
                }
                message.readBytes(bbLen);
                bbLen.limit(bbLen.capacity());
                if (bbLen.remaining() == 0) {
                    bbLen.flip();
                    int len = bbLen.getInt();
                    bbLen.clear();
                    bb = ByteBuffer.allocate(len);
                    System.out.println(len);
                }
            }
        }
    }

    public long getSessionId() {
        return sessionId;
    }

    public void setSessionId(long sessionId) {
        this.sessionId = sessionId;
        factory.addSession(sessionId, this);
    }

    public int sendResponse(Record rsp, String response) {
        byte[] bytes = "我成功了！".getBytes(StandardCharsets.UTF_8);
        ByteBuffer bb = ByteBuffer.allocate(bytes.length);
        bb.put(bytes);
        bb.flip();
        sendBuffer(bb);
        return 1;
    }

    private void sendBuffer(ByteBuffer buffers) {
        channel.writeAndFlush(Unpooled.wrappedBuffer(buffers)).addListener(onSendBufferDoneListener);
    }

    // Use a single listener instance to reduce GC
    private final GenericFutureListener<Future<Void>> onSendBufferDoneListener = f -> {
        if (f.isSuccess()) {
            System.out.println("数据包已经发送");
        }
    };
}
